package com.taobao.api.internal.tmc;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.taobao.api.internal.util.NamedThreadFactory;
import com.taobao.api.internal.util.StringUtils;
import com.taobao.api.internal.util.TaobaoUtils;
import com.taobao.api.internal.util.json.JSONReader;
import com.taobao.api.internal.util.json.JSONValidatingReader;
import com.taobao.top.link.LinkException;
import com.taobao.top.link.endpoint.EndpointContext;
import com.taobao.top.link.endpoint.Identity;

/**
 * 消息通道客户端入口。
 * 
 * @author fengsheng
 * @since 1.0, May 4, 2013
 */
public class TmcClient {

  private static final Log      log        = LogFactory.getLog(TmcClient.class);

  // data transfer protocol
  private static final String   SIZE       = "size";
  private static final String   MSG_LIST   = "msg_list";
  private static final String   SUCCESS    = "success";
  private static final String   FAIL       = "fail";
  private static final String   CONTENT    = "content";
  private static final String   SESSION    = "session";
  private static final String   TIMESTAMP  = "timestamp";
  private static final String   APP_KEY    = "app_key";
  private static final String   GROUP_NAME = "group_name";
  private static final String   SIGN       = "sign";

  protected static final String KIND       = "__kind";
  protected static final String TOPIC      = "__topic";

  // just different from type
  public class Kind {
    public static final short Data    = 0;
    public static final short Confirm = 1;
    public static final short Request = 2;
    public static final short Reply   = 3;
  }

  private final AtomicBoolean connected   = new AtomicBoolean(false);
  private final ReentrantLock fetchLock   = new ReentrantLock();

  private InnerClient         client;
  private MessageHandler      handler;

  private ThreadPoolExecutor  threadPool;
  private int                 threadCount = 20;                      // 并发处理的线程数量
  private int                 fetchSize   = 100;                     // 每次批量获取的消息数量
  private int                 fetchPeriod = 10;                      // 定时获取消息周期（单位：秒）

  private Timer               fetchTimer;
  private TimerTask           fetchTimerTask;
  private boolean             fetchStopFlag;

  private String              uri;                                   // 消息通道服务地址

  public TmcClient(String appKey, String appSecret) {
    this(appKey, appSecret, "default"); // 默认分组+线上服务
  }

  public TmcClient(String appKey, String appSecret, String groupName) {
    this("ws://mc.api.taobao.com/", appKey, appSecret, groupName); // 指定分组+线上服务
  }

  public TmcClient(String uri, String appKey, String appSecret, String groupName) {
    this.uri = uri;
    this.client = new InnerClient(new TmcIdentity(appKey, groupName));
    this.client.appKey = appKey;
    this.client.appSecret = appSecret;
    this.client.groupName = groupName;
  }

  public void setMessageHandler(MessageHandler handler) {
    this.handler = handler;
  }

  public void setThreadCount(int threadCount) {
    this.threadCount = threadCount;
  }

  public void setFetchSize(int fetchSize) {
    this.fetchSize = fetchSize;
  }

  public void setFetchPeriod(int fetchPeriod) {
    this.fetchPeriod = fetchPeriod;
  }

  public void connect() throws LinkException {
    if (!connected.compareAndSet(false, true)) {
      return;
    }

    if (this.handler != null) {
      this.client.setMessageHandler(new InnerMessageHandler());
      this.threadPool = new ThreadPoolExecutor(threadCount, threadCount, fetchPeriod, TimeUnit.MICROSECONDS, new ArrayBlockingQueue<Runnable>(fetchSize),
          new NamedThreadFactory("tmc-exe"), new CallerRunsPolicy());
    }
    this.client.connect(uri);
    this.doFetch();
  }

  /**
   * 向指定的主题发布一条与用户相关的消息。
   * 
   * @param topic
   *          主题名称
   * @param content
   *          严格根据主题定义的消息内容（JSON/XML）
   * @param session
   *          用户授权码
   */
  public void send(String topic, String content, String session) throws LinkException {
    if (StringUtils.isEmpty(content)) {
      throw new LinkException("content is required");
    }
    if (StringUtils.isEmpty(session)) {
      throw new LinkException("session is required");
    }

    Map<String, String> msg = new HashMap<String, String>();
    msg.put(KIND, Short.toString(Kind.Data));
    msg.put(TOPIC, topic);
    msg.put(CONTENT, content);
    msg.put(SESSION, session);
    this.client.sendAndWait(msg, 1000);
  }

  public void close() {
    this.stopFetch();
    if (this.threadPool != null) {
      this.threadPool.shutdown();
    }
    this.client.disconnect();
  }

  private void doFetch() {
    this.stopFetch();
    this.fetchTimerTask = new TimerTask() {
      public void run() {
        if (!fetchStopFlag && !fetchLock.isLocked()) {
          fetch();
        }
        fetchStopFlag = false;
      }
    };
    Date begin = new Date();
    begin.setTime(begin.getTime() + fetchPeriod * 1000L);
    this.fetchTimer = new Timer(true);
    this.fetchTimer.schedule(this.fetchTimerTask, begin, fetchPeriod * 1000L);
  }

  private void stopFetch() {
    if (this.fetchTimer != null) {
      this.fetchTimer.cancel();
      this.fetchTimer = null;
    }
  }

  private void fetch() {
    Map<String, String> msg = new HashMap<String, String>();
    msg.put(KIND, Short.toString(Kind.Request));
    msg.put(SIZE, Integer.toString(this.fetchSize));
    try {
      this.client.send(msg);
    } catch (LinkException e) {
      log.error("fetch message error", e);
    }
  }

  private void delayNextFetch() {
    this.fetchStopFlag = true;
  }

  class InnerClient extends MixClient {
    private String appKey;
    private String appSecret;
    private String groupName;

    public InnerClient(TmcIdentity id) {
      super(id);
    }

    @Override
    protected Map<String, String> createConnectHeaders() {
      Map<String, String> headers = new HashMap<String, String>();
      headers.put(TIMESTAMP, String.valueOf(System.currentTimeMillis()));
      headers.put(APP_KEY, this.appKey);
      headers.put(GROUP_NAME, this.groupName);
      String sign = null;
      try {
        sign = TaobaoUtils.signTopRequestNew(headers, this.appSecret, false);
      } catch (Exception e) {
        log.error("sign error", e);
      }
      headers.put(SIGN, sign);
      return headers;
    }
  }

  class InnerMessageHandler implements com.taobao.top.link.endpoint.MessageHandler {
    // on push messages
    public void onMessage(EndpointContext context) throws Exception {
      this.onMessage(context.getMessage(), context.getMessageFrom());
    }

    // on poll results
    public void onMessage(Map<String, String> message, Identity identity) {
      fetchLock.lock();
      try {
        String body = message.get(MSG_LIST);
        List<Message> msgList = parseJsonMessages(body);
        final CountDownLatch latch = new CountDownLatch(msgList.size());

        final Map<String, String> confirm = new HashMap<String, String>();
        confirm.put(KIND, Short.toString(Kind.Confirm));

        // using string buffer to make sure thread-safe
        final StringBuffer sIds = new StringBuffer();
        final StringBuffer fIds = new StringBuffer();
        for (final Message msg : msgList) {
          threadPool.submit(new Runnable() {
            public void run() {
              try {
                MessageStatus status = new MessageStatus();
                handler.onMessage(msg, status);
                if (status.isFail()) {
                  appendMsgId(fIds, msg.getId());
                } else {
                  appendMsgId(sIds, msg.getId());
                }
              } catch (Throwable e) {
                appendMsgId(fIds, msg.getId());
                log.error("handle message fail: msgId=" + msg.getId(), e);
              } finally {
                latch.countDown();
              }
            }
          });
        }

        latch.await();
        // batch confirm message handle result
        if (sIds.length() > 0) {
          confirm.put(SUCCESS, sIds.toString());
        }
        if (fIds.length() > 0) {
          confirm.put(FAIL, fIds.toString());
        }
        client.send(confirm);

        // immediately fetch messages only if confirmed
        fetch();
        delayNextFetch();
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      } catch (LinkException e) {
        log.error("confirm message error", e);
      } catch (Throwable e) {
        log.error("handle message error", e);
      } finally {
        fetchLock.unlock();
      }
    }

    private void appendMsgId(StringBuffer ids, Long id) {
      if (ids.length() > 0) {
        ids.append(",");
      }
      ids.append(id);
    }

    /**
     * 解释JSON消息成对象，消息格式：[{msg},{msg}]
     */
    private List<Message> parseJsonMessages(String json) {
      List<Message> msgList = new ArrayList<Message>();

      JSONReader reader = new JSONValidatingReader();
      Object root = reader.read(json);
      if (root instanceof List<?>) {
        List<?> list = (List<?>) root;
        for (Object obj : list) {
          if (obj instanceof Map<?, ?>) {
            Map<?, ?> map = (Map<?, ?>) obj;
            msgList.add(Message.parse(map));
          }
        }
      }

      return msgList;
    }
  }

}
